package test.consumer;

import javax.jms.*;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import com.prm.excelserver.protobuf.*;
import java.util.concurrent.CountDownLatch;

import com.google.protobuf.*;

public class Consumer {
    // URL of the JMS server
    //private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	private static String url = "failover://(tcp://119.247.167.227:61616)";

    // Name of the queue we will receive messages from
    private static String subjectInst = "testMqQueueInstrument";
    private static String subjectTrade = "testMqQueueTrade";

    public static void main(String[] args) throws JMSException {
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(2);   	
    	
        //ExcelServer tradeServer = new ExcelServer("prm", "Trade", tradeItems, latch);
        
    	// Getting JMS connection from the server
        ConnectionFactory connectionFactory
            = new ActiveMQConnectionFactory(url);
        Connection connection = connectionFactory.createConnection();
        connection.start();

        // Creating session for seding messages
        Session session = connection.createSession(false,
            Session.AUTO_ACKNOWLEDGE);

        // Getting the queue 'TESTQUEUE'
        //Destination destination = session.createQueue(subject);
        Destination destInst = session.createTopic(subjectInst);
        Destination destTrade = session.createTopic(subjectTrade);

        // MessageConsumer is used for receiving (consuming) messages
        MessageConsumer consumerInst = session.createConsumer(destInst);
        MessageConsumer consumerTrade = session.createConsumer(destTrade);
 
        // Here we receive the message.
        // By default this call is blocking, which means it will wait
        // for a message to arrive on the queue.
        RxThread instRun = new RxThread(consumerInst, "Instrument", startSignal, doneSignal);
        RxThread tradeRun = new RxThread(consumerTrade, "Trade", startSignal, doneSignal);
        
    	Runtime.getRuntime().addShutdownHook(new Thread() {
    		@Override
    		public void run() {
    			System.out.println("System terminates..");
    			RxThread.stopRunning();
    		}
    	});
        
        Thread instThr = new Thread(instRun);
        Thread tradeThr = new Thread(tradeRun);
        instThr.start();
        tradeThr.start();
        startSignal.countDown();
        try {
        	doneSignal.await();
        } catch (InterruptedException e) {
        	System.out.println("in Consumer class: donesignal.wait() error");
        }
        
        connection.close();
       
        
    }
}